package com.kongxiang.storm;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.spout.SchemeAsMultiScheme;
import backtype.storm.topology.TopologyBuilder;
import com.kongxiang.storm.bolt.*;
import com.kongxiang.storm.constant.Descid;
import com.kongxiang.storm.constant.LogTableName;
import com.kongxiang.storm.spout.PWSpout;
import org.apache.storm.jdbc.bolt.JdbcInsertBolt;
import storm.kafka.KafkaSpout;
import storm.kafka.SpoutConfig;
import storm.kafka.StringScheme;
import storm.kafka.ZkHosts;
import java.util.ArrayList;
import java.util.List;

public class KafkaTopology {
    public static void main(String[] args) throws InterruptedException {
        System.setProperty("java.net.preferIPv6Addresses", "false");
        String brokerZkStr = "127.0.0.1:2181";
        String brokerZkPath = "/brokers";
        //消费kafka得top
        String topic = "demo";
        String offset = "";
        //id 可以随意命名
        String id = "demo";
        Integer workerNumSpout = 1;
        Integer workerNumBolt = 1;
        Integer maxSpoutPending = 2000;

        if(args.length > 1){
            topic = args[1];
        }
        if(args.length > 2){
            workerNumSpout = Integer.parseInt(args[2]);
            workerNumBolt = Integer.parseInt(args[3]);
        }
        if(args.length > 4){
            maxSpoutPending = Integer.parseInt(args[4]);
        }
        ZkHosts zk = new ZkHosts(brokerZkStr,brokerZkPath);
        //消费配置
        SpoutConfig spoutConf = new SpoutConfig(zk, topic,offset,id);
        List<String> zkServices = new ArrayList<String>();
        zkServices.add(brokerZkStr);
        spoutConf.zkServers = zkServices;
        spoutConf.zkPort = 2181;
        spoutConf.forceFromStart = false;
        spoutConf.socketTimeoutMs = 60 * 1000;
        spoutConf.scheme = new SchemeAsMultiScheme(new StringScheme());
        TopologyBuilder builder = new TopologyBuilder();
        // Kafka我们创建了一个5分区的Topic，这里并行度设置为5
        builder.setSpout("data", new KafkaSpout(spoutConf), workerNumSpout);
        // 数据源
//        builder.setSpout("data", new PWSpout(), 1);
        // 随机分组
        builder.setBolt("analyze", new DistributeBolt(), 3).shuffleGrouping("data");

        builder.setBolt(Descid.REQUEST_BOLT_ID.name(), new DealRequestBolt(), 1) .shuffleGrouping(
            "analyze",// --- 发送方名字
            Descid.REQUEST_STREAM_ID.name()  // --- 接收发送方该stream 的tuple
        );

        builder.setBolt(Descid.EXCEPTION_BOLT_ID.name(), new DealExceptionBolt(), 1) .shuffleGrouping(
            "analyze",// --- 发送方名字
            Descid.EXCEPTION_STREAM_ID.name()  // --- 接收发送方该stream 的tuple
        );
        builder.setBolt(Descid.OTHER_BOLT_ID.name(), new OtherBolt(), 1) .shuffleGrouping(
            "analyze",// --- 发送方名字
            Descid.OTHER_STREAM_ID.name()  // --- 接收发送方该stream 的tuple
        );
        // 插入Other 数据 分类插入数据
        JdbcInsertBolt jdbcInsertBolt = JDBCBolt.getJdbcInsertBolt(LogTableName.other_log.name());
        builder.setBolt(Descid.INSERT_OTHER_BOLT_ID.name(), jdbcInsertBolt, 1) .shuffleGrouping(Descid.OTHER_BOLT_ID.name());

        JdbcInsertBolt requestInsertBolt = JDBCBolt.getJdbcInsertBolt(LogTableName.request_log.name());
        builder.setBolt(Descid.INSERT_REQUEST_BOLT_ID.name(), requestInsertBolt, 1) .shuffleGrouping(Descid.REQUEST_BOLT_ID.name());

        JdbcInsertBolt exceptionInsertBolt = JDBCBolt.getJdbcInsertBolt(LogTableName.exception_log.name());
        builder.setBolt(Descid.INSERT_EXCEPTION_BOLT_ID.name(), exceptionInsertBolt, 1) .shuffleGrouping(Descid.EXCEPTION_BOLT_ID.name());
        // END
        Config config = new Config();
        config.setDebug(false);
        config.setNumWorkers(workerNumSpout);
        config.setMaxSpoutPending(1);
        config.setNumAckers(0);
        config.setDebug(false);
        if(maxSpoutPending > 0){
            config.setMaxSpoutPending(maxSpoutPending);
        }
        if(args.length>0){
            try {
                // args有参数时在分布式上提交任务
                StormSubmitter.submitTopology(args[0], config, builder.createTopology());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }else{
            // args没有参数时在本地提交任务
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("KafkaTop", config, builder.createTopology());
        }
    }
}
